package com.gitee.xhs.keyedState.mapState;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/6/30 16:27
 * @description MapStateKeyedProcessFunction
 */
public class MapStateKeyedProcessFunction extends KeyedProcessFunction<Long, Transaction, String> {

    private transient MapState<Integer, Double> mapState;

    @Override
    public void open(Configuration configuration) {
        mapState = getRuntimeContext().getMapState(MapStateConstant.MAP_STATE_DESCRIPTOR);
    }

    @Override
    public void processElement(Transaction value, KeyedProcessFunction<Long, Transaction, String>.Context ctx, Collector<String> out) throws Exception {

        int maxIndex = 0;
        for (Integer index : mapState.keys()) {
            if (maxIndex < index) {
                maxIndex = index;
            }
        }
        maxIndex += 1;
        mapState.put(
                maxIndex,
                value.getAmount()
        );

        out.collect(value.toString());
    }
}
